
插入Python指令。在影刀流程编辑器中,从左侧指令面板拖入"Python脚本"指令。支持两种模式:简单脚本(直接写代码)和模块编程(创建独立的Python模块,可复用)。简单场景用简单脚本,复杂逻辑用模块编程。
变量传递。可视化流程中的变量可以传入Python脚本(在指令配置中映射),Python脚本的输出也可以存回流程变量。注意:Python中修改变量不会自动同步回流程,必须通过"输出变量"显式返回。
环境和依赖。影刀内置Python运行环境,常用库(pandas、requests、openpyxl、re、json等)已预装。需要额外的库可以在影刀的Python环境中pip安装。建议用虚拟环境隔离,避免依赖冲突。
用法一:pandas批量处理Excel。10万行数据用可视化循环要半小时,用pandas只要3秒。读取Excel→数据清洗(去重、填充、格式转换)→分组聚合→多表合并→输出结果,全部在一个Python指令里完成。
用法二:requests调用复杂API。遇到需要签名、Token刷新、重试机制的API,用requests+自定义函数搞定。支持GET/POST/PUT/DELETE,自动处理Cookie和Session,失败自动重试3次,比可视化的HTTP指令灵活10倍。
用法三:正则+文本处理。从一堆杂乱文本中提取邮箱、手机号、身份证号、订单号,用正则表达式一行搞定。还可以做文本清洗(去HTML标签、去特殊字符、标准化格式)、关键词提取、相似度计算。
用法四:文件批量操作。批量重命名、批量转换格式、批量压缩解压、批量查找替换。用os和shutil库,几十行代码搞定几百个文件的操作,比可视化循环快得多,也不容易出错。
用法五:自定义算法和AI集成。需要做数据预测、图片识别、文本分类时,在Python里调用sklearn、PIL、transformers等库。也可以调用OpenAI/DeepSeek API做智能处理,把AI能力嵌入RPA流程。
import pandas as pd
import os
# 1. 读取多个Excel并合并
input_dir = r'C:\data\input'
files = [f for f in os.listdir(input_dir) if f.endswith('.xlsx')]
dfs = []
for f in files:
df = pd.read_excel(os.path.join(input_dir, f))
df['来源文件'] = f
dfs.append(df)
merged = pd.concat(dfs, ignore_index=True)
# 2. 数据清洗
merged = merged.drop_duplicates() # 去重
merged['金额'] = pd.to_numeric(merged['金额'], errors='coerce').fillna(0)
merged['日期'] = pd.to_datetime(merged['日期'], errors='coerce')
# 3. 按区域分组汇总
summary = merged.groupby('区域').agg({
'金额': ['sum', 'mean', 'count'],
'订单号': 'nunique'
}).reset_index()
summary.columns = ['区域', '总金额', '平均金额', '订单数', '客户数']
# 4. 输出
output_path = r'C:\data\output\汇总结果.xlsx'
with pd.ExcelWriter(output_path) as writer:
merged.to_excel(writer, sheet_name='明细', index=False)
summary.to_excel(writer, sheet_name='汇总', index=False)
print(f'处理完成,共{len(merged)}条明细,{len(summary)}个区域')
import requests
import time
import json
def call_api(url, method='GET', headers=None, data=None, max_retries=3):
"""带重试的API调用"""
for attempt in range(max_retries):
try:
if method.upper() == 'GET':
resp = requests.get(url, headers=headers, params=data, timeout=30)
else:
resp = requests.post(url, headers=headers, json=data, timeout=30)
resp.raise_for_status()
return resp.json()
except Exception as e:
print(f'第{attempt+1}次调用失败: {e}')
if attempt < max_retries - 1:
time.sleep(2 ** attempt) # 指数退避
return None
# 使用示例
result = call_api(
'https://api.example.com/data',
method='POST',
headers={'Authorization': 'Bearer YOUR_TOKEN', 'Content-Type': 'application/json'},
data={'keyword': 'test', 'page': 1}
)
if result:
print(json.dumps(result, ensure_ascii=False, indent=2))